package analyse;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.processor.Processor;
import org.apache.kafka.streams.processor.ProcessorContext;
import org.apache.kafka.streams.processor.PunctuationType;

import java.time.Duration;
import java.util.*;
import java.util.Map.Entry;

/**
 * 使用Stream实现解析器
 * 采样间隔通过流操作自身的窗口功能实现
 *
 * @author pf
 */
@Deprecated
public class SamplerAnalyseStream implements Runnable {

	private String brokerUrl;
	private String topic;
	private int interval;
	public SamplerAnalyseStream(String brokerUrl, String topic, int interval) {

		this.brokerUrl = brokerUrl;
		this.topic = topic;
		this.interval = interval;
	}

	public static void main(String[] args) {

		new Thread(new SamplerAnalyseStream("172.18.84.143:9092", "test", 10)).start();
	}

	/**
	 * @param brokerUrl kafka连接地址
	 * @param topic     topic名称
	 * @param interval  采样间隔，单位：秒
	 */
	public void run() {

		Properties props = new Properties();

		props.put(StreamsConfig.APPLICATION_ID_CONFIG, UUID.randomUUID().toString());
		props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
		props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
		props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

		StreamsBuilder builder = new StreamsBuilder();

		KStream<String, String> textLines = builder.stream(topic);

		textLines.process(() -> new MyProcess(interval));

		KafkaStreams streams = new KafkaStreams(builder.build(), props);
		streams.start();
	}
}

@Deprecated
class MyProcess implements Processor<String, String> {

	HashMap<String, ArrayList<JSONObject>> samplers = new HashMap<String, ArrayList<JSONObject>>();
	private ProcessorContext context;
	private int interval;

	public MyProcess(int interval) {

		this.interval = interval;
	}

	@Override
	public void init(ProcessorContext context) {

		this.context = context;
		this.context.schedule(Duration.ofSeconds(interval), PunctuationType.STREAM_TIME, (timeStamp) -> {
			long start = System.currentTimeMillis();
			JSONArray res = analyse(timeStamp);
			System.out.println((System.currentTimeMillis() - start) + res.toJSONString());
		});
	}


	@Override
	public void process(String key, String value) {

//		System.out.println(value);

		JSONObject valueJson = JSONObject.parseObject(value);

		ArrayList<JSONObject> tempList = samplers.get(key);
		if (tempList == null) {
			tempList = new ArrayList<JSONObject>();
		}

		tempList.add(valueJson);
		samplers.put(key, tempList);
	}

	@Override
	public void close() {
	}

	public JSONArray analyse(long timeStamp) {

		JSONArray resArry = new JSONArray();

		JSONObject total = new JSONObject();

		double TOTAL_sendBytes = 0.0;
		double TOTAL_receiveBytes = 0.0;
		int TOTAL_threadCount = 0;
		int TOTAL_successCount = 0;
		int TOTAL_failCount = 0;
		int TOTAL_totalDelay = 0;
		int TOTAL_avgDelay = 0;
		int TOTAL_qps = 0;
		int TOTAL_length = 0;
		int TOTAL_length_level1 = 0;
		int TOTAL_length_level2 = 0;
		int TOTAL_length_level3 = 0;
		ArrayList<Integer> TOTAL_delays = new ArrayList<Integer>();

		for (Entry<String, ArrayList<JSONObject>> entry : samplers.entrySet()) {

			String key = entry.getKey();
			ArrayList<JSONObject> value = entry.getValue();

			JSONObject resObj = new JSONObject();

			double sendBytes = 0.0;
			double receiveBytes = 0.0;
			int threadCount = 0;
			int successCount = 0;
			int failCount = 0;
			int totalDelay = 0;
			int avgDelay = 0;
			int qps = 0;
			int length = value.size();
			int length_level1 = length * 90 / 100 - 1;
			int length_level2 = length * 95 / 100 - 1;
			int length_level3 = length * 99 / 100 - 1;

			if (length_level1 <= 0) {
				continue;
			}

			ArrayList<Integer> delays = new ArrayList<Integer>();

			for (JSONObject jsonObject : value) {

				// 计算这段时间内的采样数
				if (jsonObject.getBooleanValue("success")) {
					successCount = successCount + jsonObject.getIntValue("sampleCount");
				} else {
					failCount = failCount + jsonObject.getIntValue("sampleCount");
				}

				// 计算这段时间内的总耗时
				int delay = jsonObject.getIntValue("delayMS");
				totalDelay = totalDelay + delay;
				delays.add(delay);

				// 获取这段时间内并发用户数的最大值
				int tempThread = jsonObject.getIntValue("threadCount");
				if (threadCount < tempThread) {
					threadCount = tempThread;
				}

				// 计算这段时间内总的发送MB
				sendBytes = sendBytes + jsonObject.getDoubleValue("sentBytes");

				// 计算这段时间内总的接受MB
				receiveBytes = receiveBytes + jsonObject.getDoubleValue("receiveBytes");
			}

			if (successCount + failCount <= 0) {
				System.out.println("");
				continue;
			}

			Collections.sort(delays);

			avgDelay = totalDelay / (successCount + failCount);
			qps = (successCount + failCount) / interval;

			resObj.put("timeStamp", timeStamp);
			resObj.put("label", key);
			resObj.put("successCount", successCount);
			resObj.put("failCount", failCount);
			resObj.put("avgTime", avgDelay);
			resObj.put("qps", qps);
			resObj.put("90%Delay", delays.get(length_level1));
			resObj.put("95%Delay", delays.get(length_level2));
			resObj.put("99%Delay", delays.get(length_level3));
			resObj.put("minDelay", delays.get(0));
			resObj.put("maxDelay", delays.get(length - 1));
			resObj.put("sendBytes/s", sendBytes / interval);
			resObj.put("receiveBytes/s", receiveBytes / interval);
			resObj.put("thread", threadCount);

			resArry.add(resObj);

			// 添加到TOTAL计数中
			TOTAL_successCount = TOTAL_successCount + successCount;
			TOTAL_failCount = TOTAL_failCount + failCount;
			TOTAL_totalDelay = TOTAL_totalDelay + totalDelay;
			TOTAL_delays.addAll(delays);
			TOTAL_sendBytes = TOTAL_sendBytes + sendBytes;
			TOTAL_receiveBytes = TOTAL_receiveBytes + receiveBytes;
			if (TOTAL_threadCount < threadCount) {
				TOTAL_threadCount = threadCount;
			}

			TOTAL_length = TOTAL_length + length;
		}

		TOTAL_length_level1 = TOTAL_length * 90 / 100 - 1;
		TOTAL_length_level2 = TOTAL_length * 95 / 100 - 1;
		TOTAL_length_level3 = TOTAL_length * 99 / 100 - 1;

		if (TOTAL_length_level1 > 0) {
			Collections.sort(TOTAL_delays);

			TOTAL_avgDelay = TOTAL_totalDelay / (TOTAL_successCount + TOTAL_failCount);
			TOTAL_qps = (TOTAL_successCount + TOTAL_failCount) / interval;

			total.put("timeStamp", timeStamp);
			total.put("label", "Total");
			total.put("successCount", TOTAL_successCount);
			total.put("failCount", TOTAL_failCount);
			total.put("avgTime", TOTAL_avgDelay);
			total.put("qps", TOTAL_qps);
			total.put("90%Delay", TOTAL_delays.get(TOTAL_length_level1));
			total.put("95%Delay", TOTAL_delays.get(TOTAL_length_level2));
			total.put("99%Delay", TOTAL_delays.get(TOTAL_length_level3));
			total.put("minDelay", TOTAL_delays.get(0));
			total.put("maxDelay", TOTAL_delays.get(TOTAL_length - 1));
			total.put("sendBytes/s", TOTAL_sendBytes / interval);
			total.put("receiveBytes/s", TOTAL_receiveBytes / interval);
			total.put("thread", TOTAL_threadCount);

			resArry.add(total);
		}

		samplers.clear();

		return resArry;
	}

}
